package org.study.module.hadoop;

import org.study.module.common.AbstractConfigBuilder;

/**
 * YARN配置文件构建
 * yarn-site.xml
 */
public class YARNConfigBuilder extends AbstractConfigBuilder {

    /**
     * 客户端提交作业的 RPC 地址（默认 8032 端口）
     *
     * @param address Resource地址
     */
    public YARNConfigBuilder resourceManagerAddress(String address) {
        addMessage("客户端提交作业的 RPC 地址");
        addProperty("yarn.resourcemanager.address", address);
        return this;
    }

    /**
     * ApplicationMaster 申请资源的调度器地址（默认 8030 端口）
     *
     * @param schedulerAddress 调度地址
     */
    public YARNConfigBuilder resourceManagerSchedulerAddress(String schedulerAddress) {
        addMessage("ApplicationMaster 申请资源的调度器地址");
        addProperty("yarn.resourcemanager.scheduler.address", schedulerAddress);
        return this;
    }

    /**
     * ResourceManager Web UI 地址（默认 8088 端口）
     *
     * @param webappAddress 调度地址
     */
    public YARNConfigBuilder resourceManagerWebappAddress(String webappAddress) {
        addMessage("ResourceManager Web UI 地址");
        addProperty("yarn.resourcemanager.webapp.address", webappAddress);
        return this;
    }

    /**
     * 指定 NodeManager 向 ResourceManager 注册和发送心跳的 RPC 服务地址
     * 处理 NodeManager 的资源汇报（如 Container 状态、节点健康信息等）
     * 非HA模式， 默认端口 8031
     *
     * @param trackerAddress 格式为 host:port
     */
    public YARNConfigBuilder resourceManagerResourceTrackerAddress(String trackerAddress) {
        addMessage("注册和发送心跳的 RPC 服务地址-->");
        addProperty("yarn.resourcemanager.resource-tracker.address", trackerAddress);
        return this;
    }

    /**
     * 在非HA模式下使用，ResourceManager
     *
     * @param hostname RM的hostname
     */
    public YARNConfigBuilder resourceManagerHostname(String hostname) {
        addMessage("RM的hostname");
        addProperty("yarn.resourcemanager.hostname", hostname);
        return this;
    }

    /**
     * 管理员与 ResourceManager 通信的 RPC 地址
     * 在非HA模式下使用，ResourceManager,默认端口 8033
     * 指定管理员向 ResourceManager 发送管理命令（如刷新队列、节点列表等）的 RPC 服务地址
     *
     * @param adminAddress RM的
     */
    public YARNConfigBuilder resourceManagerAdminAddress(String adminAddress) {
        addMessage("管理员与 ResourceManager 通信的 RPC 地址");
        addProperty("yarn.resourcemanager.admin.address", adminAddress);
        return this;
    }

    /**
     * 指定YARN集群ID（逻辑名称，所有RM共享)
     *
     * @param clusterId 集群id
     */
    public YARNConfigBuilder resourceManagerClusterId(String clusterId) {
        addMessage("YARN集群的唯一标识，用于区分不同集群");
        addProperty("yarn.resourcemanager.cluster-id", clusterId);
        return this;
    }

    /**
     * resourceManager 启用 HA 模式
     */
    public YARNConfigBuilder resourceManagerHa() {
        return resourceManagerHa("true");
    }

    /**
     * resourceManager 启用 HA 模式
     *
     * @param enabled ture,false
     */
    public YARNConfigBuilder resourceManagerHa(String enabled) {
        addMessage("启用 RM HA 模式");
        addProperty("yarn.resourcemanager.ha.enabled", enabled);
        return this;
    }

    /**
     * 配置各 RM 主机地址
     *
     * @param rmId   rm节点的id
     * @param rmHost rm节点的hostname
     */
    public YARNConfigBuilder haResourceMangerHostName(String rmId, String rmHost) {
        addMessage("配置各 RM 主机地址"+rmId);
        addProperty("yarn.resourcemanager.hostname." + rmId, rmHost);
        return this;
    }


    /**
     * HA模式 定义 ResourceManager 节点
     *
     * @param rmIds rm节点
     */
    public YARNConfigBuilder resourceManagerHaRmIds(String rmIds) {
        addMessage("RM集群ID");
        addProperty("yarn.resourcemanager.ha.rm-ids", rmIds);
        return this;
    }

    /**
     * 指定 NodeManager 向 ResourceManager 注册和发送心跳的 RPC 服务地址
     * 处理 NodeManager 的资源汇报（如 Container 状态、节点健康信息等）
     * HA模式 默认端口 8031
     *
     * @param rmId           RM节点id
     * @param trackerAddress 格式为 host:port
     */
    public YARNConfigBuilder haResourceManagerResourceTrackerAddress(String rmId, String trackerAddress) {
        addMessage("注册和发送心跳的 RPC 服务地址"+rmId);
        addProperty("yarn.resourcemanager.resource-tracker.address." + rmId, trackerAddress);
        return this;
    }

    /**
     * Resource Manager 调度器地址
     * 默认端口：8030
     *
     * @param rmId             RM节点id
     * @param schedulerAddress 格式为 host:port
     */
    public YARNConfigBuilder haResourceManagerSchedulerAddress(String rmId, String schedulerAddress) {
        addMessage("Resource Manager 调度器地址"+rmId);
        addProperty("yarn.resourcemanager.scheduler.address." + rmId, schedulerAddress);
        return this;
    }

    /**
     * 客户端提交作业的 RPC 地址（默认 8032 端口）
     *
     * @param address Resource地址
     */
    public YARNConfigBuilder haResourceManagerAddress(String rmId, String address) {
        addMessage("客户端提交作业的 RPC 地址"+rmId);
        addProperty("yarn.resourcemanager.address." + rmId, address);
        return this;
    }

    /**
     * 管理员与 ResourceManager 通信的 RPC 地址
     * 在HA模式下使用，ResourceManager,默认端口 8033
     * 指定管理员向 ResourceManager 发送管理命令（如刷新队列、节点列表等）的 RPC 服务地址
     *
     * @param rmId         RM的节点id
     * @param adminAddress RM的
     */
    public YARNConfigBuilder haResourceManagerAdminAddress(String rmId, String adminAddress) {
        addMessage("管理员与 ResourceManager 通信的 RPC 地址"+rmId);
        addProperty("yarn.resourcemanager.admin.address." + rmId, adminAddress);
        return this;
    }

    /**
     * ResourceManager Web UI 地址（默认 8088 端口）
     *
     * @param webappAddress 调度地址
     */
    public YARNConfigBuilder haResourceManagerWebappAddress(String rmId, String webappAddress) {
        addMessage("ResourceManager Web UI 地址"+rmId);
        addProperty("yarn.resourcemanager.webapp.address." + rmId, webappAddress);
        return this;
    }

    /**
     * ResourceManager HTTPS Web UI地址（若启用SSL）默认端口 8090
     *
     * @param webappAddress HTTPS Web UI地址
     */
    public YARNConfigBuilder haResourceManagerWebappHttpsAddress(String rmId, String webappAddress) {
        addMessage("ResourceManager Web UI 地址"+rmId);
        addProperty("yarn.resourcemanager.webapp.https.address." + rmId, webappAddress);
        return this;
    }

    /**
     * HA模式  ResourceManager Zookeeper地址,多个用逗号拼接，端口默认2181  如：host1:2181,host2:2181
     *
     * @param zkAddress Zookeeper地址
     */
    public YARNConfigBuilder resourceManagerZkAddress(String zkAddress) {
        addMessage("Zookeeper地址");
        addProperty("yarn.resourcemanager.zk-address", zkAddress);
        addMessage("ResourceManager 启用RM自动故障转移（依赖ZooKeeper）");
        addProperty("yarn.resourcemanager.ha.automatic-failover.enabled", "true");
        addMessage("ResourceManager 使用ZooKeeper存储RM状态（替代默认的FileSystemRMStateStore）");
        resourceManagerStoreClass("org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore");
        return this;
    }

    /**
     * Kerberos  resourceManager,添加认证
     */
    public YARNConfigBuilder resourceManagerPrincipal(String installPath, String host, String realm) {
        addMessage("ResourceManager 添加 Kerberos 认证-->");
        addProperty("yarn.resourcemanager.principal", "yarn/" + host + "@" + realm);
        addMessage("ResourceManager 认证文件路径-->");
        addProperty("yarn.resourcemanager.keytab.file", HadoopShell.getHadoopKerberos(installPath) + "/yarn-" + host + ".keytab");
        addMessage("ResourceManager 启用YARN服务认证-->");
        addProperty("yarn.resourcemanager.security.enable", "true");
        return this;
    }

    /**
     * Kerberos  nodeManager,添加认证
     */
    public YARNConfigBuilder nodeManagerPrincipal(String path, String host, String realm) {
        addMessage("NodeManager 添加 Kerberos 认证");
        addProperty("yarn.nodemanager.principal", "yarn/" + host + "@" + realm);
        addMessage("NodeManager 认证文件路径");
        addProperty("yarn.nodemanager.keytab.file", path + "/hdfs-" + host + ".keytab");
        addMessage("NodeManager 启用YARN服务认证");
        addProperty("yarn.nodemanager.security.enable", "true");
        return this;
    }

    /**
     * HA模式，启用自动恢复
     */
    public YARNConfigBuilder resourceManagerRecovery() {
        return resourceManagerRecovery("true");
    }

    /**
     * HA模式，启用自动恢复
     *
     * @param enabled true,false
     */
    public YARNConfigBuilder resourceManagerRecovery(String enabled) {
        addMessage("启用自动恢复");
        addProperty("yarn.resourcemanager.recovery.enabled", enabled);
        return this;
    }

    /**
     * HA模式，RM状态存储类，
     * 类：org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore
     */
    public YARNConfigBuilder resourceManagerStoreClass() {
        return resourceManagerStoreClass("org.apache.hadoop.yarn.server.resourcemanager.recovery.ZKRMStateStore");
    }

    /**
     * HA模式， RM状态存储类
     *
     * @param storeClass 存储实现类
     */
    public YARNConfigBuilder resourceManagerStoreClass(String storeClass) {
        addMessage("RM状态存储类");
        addProperty("yarn.resourcemanager.store.class", storeClass);
        return this;
    }

    /**
     * 指定 NodeManager 的辅助服务（MapReduce 需设为 mapreduce_shuffle）
     */
    public YARNConfigBuilder nodeManagerAuxServices() {
        nodeManagerAuxServices("mapreduce_shuffle");
        addMessage("shuffle服务的实现类");
        addProperty("yarn.nodemanager.aux-services.mapreduce_shuffle.class", "org.apache.hadoop.mapred.ShuffleHandler");
        return this;
    }

    /**
     * 指定 NodeManager 的辅助服务（MapReduce 需设为 mapreduce_shuffle）
     *
     * @param auxServices 辅助服务
     */
    public YARNConfigBuilder nodeManagerAuxServices(String auxServices) {
        addMessage("指定 NodeManager 的辅助服务");
        addProperty("yarn.nodemanager.aux-services", auxServices);
        return this;
    }

    /**
     * NodeManager RPC 地址（默认 localhost:8041）
     *
     * @param address RPC 地址
     */
    public YARNConfigBuilder nodeManagerAddress(String address) {
        addMessage("NodeManager RPC 地址");
        addProperty("yarn.nodemanager.address", address);
        return this;
    }

    /**
     * 容器中间数据存储目录（多路径逗号分隔）默认 /tmp
     *
     * @param localDirs 中间数据存储目录
     */
    public YARNConfigBuilder nodeManagerLocalDirs(String localDirs) {
        addMessage("NodeManager RPC 地址");
        addProperty("yarn.nodemanager.local-dirs", localDirs);
        return this;
    }

    /**
     * 容器日志存储目录 （多路径逗号分隔）默认 /tmp/log
     *
     * @param logDirs 日志存储目录
     */
    public YARNConfigBuilder nodeManagerLogDirs(String logDirs) {
        addMessage("NodeManager RPC 地址");
        addProperty("yarn.nodemanager.log-dirs", logDirs);
        return this;
    }

    /**
     * 节点总可用物理内存（默认 8GB），需根据机器内存调整（如 64GB 机器设为 50GB）。
     *
     * @param resourceMemory 可用内存， 单位MB
     * @return
     */
    public YARNConfigBuilder nodeManagerResourceMemory(String resourceMemory) {
        addMessage("NodeManager 节点总可用物理内存");
        addProperty("yarn.nodemanager.resource.memory-mb", resourceMemory);
        return this;
    }

    /**
     * 节点总虚拟 CPU 核数（默认 8），建议设为物理核数的 1~2 倍。
     *
     * @param resourceCpu 总虚拟 CPU 核数
     */
    public YARNConfigBuilder nodeManagerResourceCpu(String resourceCpu) {
        addMessage("NodeManager 节点总虚拟 CPU 核数");
        addProperty("yarn.nodemanager.resource.cpu-vcores", resourceCpu);
        return this;
    }

    public void addConfigHead() {
        configs.add("cat > $HADOOP_HOME/etc/hadoop/yarn-site.xml << EOF");
        super.addConfigHead();
    }
}
